package com.adingxiong.cft.config;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.PartitionOffset;
import org.springframework.kafka.annotation.TopicPartition;
import org.springframework.kafka.listener.ConsumerAwareListenerErrorHandler;
import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * @ClassName KafkaConsumer
 * @Description  监听消息服务器的消息   可以指定 主题,消费者id, 主题的分区和多少条
 * @Author xiongchao
 * @Date 2021/3/24 13:31
 **/
@Component
@Slf4j
public class KafkaConsumer {

    @KafkaListener(topics = {"test","topic2","topic3","auto-topic"})
    public void onMessage(ConsumerRecord<?,?> record) {
        log.info("自动监听服务器消息: topic为:{} ,分区数:{} ,值:{}",record.topic(),record.partition(),record.value());
    }


    /**
     *  配置说明 :
     *      消费者id: consumer1
     *      groupid: test-group
     *      topic: 主题
     *      topicPartitions : 更详细的说明  分区数据 和offset数
     *
     *      监听topic2的0号分区，同时监听topic2的0号分区和topic2的1号分区里面offset从8开始的消息。
     * @param record
     */
    @KafkaListener(id="consumer1" ,groupId = "test-group", topicPartitions =  {
            @TopicPartition(topic = "topic3",partitions = { "0"}),
            @TopicPartition(topic = "topic2",partitions = "0", partitionOffsets = @PartitionOffset(partition = "1" ,initialOffset = "8"))
    })
    public void onMessage2(ConsumerRecord<?,?> record) {
        log.info("自动监听服务器消息: topic为:{} ,分区数:{} ,值:{} ,时间:{}",record.topic(),record.partition(),record.value(),record.timestamp());
    }


    @KafkaListener(id = "consumer2",groupId = "felix-group", topics = "topic2")
    public void onMessage3(List<ConsumerRecord<?, ?>> records) {
        System.out.println(">>>批量消费一次，records.size()="+records.size());
        for (ConsumerRecord<?, ?> record : records) {
            System.out.println(record.value());
        }
    }
    /**
     * 将这个异常处理器的BeanName放到@KafkaListener注解的errorHandler属性里面
      */
    @KafkaListener(topics = {"topic2"},errorHandler = "consumerAwareErrorHandler")
    public void onMessage4(ConsumerRecord<?, ?> record) throws Exception {
        throw new Exception("简单消费-模拟异常");
    }
    /**
     * 自定义异常  用于捕捉消费过程中出现的问题
     * @return
     */
    @Bean
    public ConsumerAwareListenerErrorHandler  consumerAwareErrorHandler() {
        return ((message, exception, consumer) -> {
            log.error("消费异常:{}",message.getPayload());
            return  null;
        });
    }

    /**
     * 消息转发
     */

    @KafkaListener(topics = "{topic1}")
    @SendTo("topic2")
    public void onMessage5(ConsumerRecord<?,?> record) {
        log.info("转发消息topic: {},内容为:{}",record.topic(),record.value());
    }


}
